/*----------------------------------------------------------------------------------------
*
*  Copyright 2019, Gao Hai Hui, <fromzeropoint@126.com>.  All rights reserved.
*  https://gitee.com/helloworldghh/xoskit.git
*  Use of this source code is governed by a MIT license
*  that can be found in the License file.
*
----------------------------------------------------------------------------------------*/
#include "../import/head.h"
#include "../data_struct/head.h"
#include "../macro/head.h"
#include "aio_helper.h"
#include "aio_thread.h"
#include "head.h"

/////////////////////////////////////////////////////////////////////////////////////////
// 
namespace xos_aio
{
    aio_thread::aio_thread( aio_impl * pAioImpl ) : m_pAioImpl( pAioImpl )
    {
        m_pThread = 0;
    }

    aio_thread::~aio_thread()
    {
    }

    ////////////////////////////////////////////////////////////////////////////////////
    // 

    int aio_thread::notify( msg_impl * p )
    {
        int nRet = 0;
        return nRet;
    }

    int aio_thread::init()
    {
        int nRet = 0;

        if( 0 == nRet )
        {
            m_pThread = mgr::container()->xos()->thread();
        }

        if( 0 == nRet )
        {
            m_pThread->create( aio_thread::thread, this );
        }

        return nRet;
    }

    int aio_thread::term()
    {
        int nRet = 0;

        if( m_pThread )
        {
            m_pThread->wait();
            m_pThread->release();
            m_pThread = 0;
        }

        m_pAioImpl = 0;

        return nRet;
    }

    ////////////////////////////////////////////////////////////////////////////////////
    // 

    void * aio_thread::thread( void * lpData )
    {
        aio_thread & obj = *( aio_thread * )lpData;
        int nRet = 0;

        while( 0 == nRet )
        {
            nRet = obj.run();
            obj.proc_close();
        }

        return 0;
    }

    int aio_thread::run()
    {
        using namespace std;
        int nRet = 0;

        OverlappedEx * lpOverlappedEx = 0;
        aio_key * lpAioKey = 0;
        aio_data * lpAioData = 0;
        DWORD dwBytes = 0;
        DWORD dwErr = 0;
        BOOL bRet = TRUE;

        if( 0 == nRet )
        {
            bRet = GetQueuedCompletionStatus( m_pAioImpl->m_hIocp, &dwBytes, ( ULONG_PTR * )&lpAioKey, ( LPOVERLAPPED * )&lpOverlappedEx, INFINITE );
        }

        if( ( 0 == nRet ) && !bRet )
        {
            dwErr = GetLastError();
            if( ERROR_ABANDONED_WAIT_0 == dwErr )
            {
                nRet = 1;
            }
            else if( ERROR_OPERATION_ABORTED == dwErr )
            {
                // CancelIo called, 正常处理就行了
            }
            else if( !lpOverlappedEx )
            {
                nRet = 2;
            }
        }

        if( ( 0 == nRet ) && lpOverlappedEx )
        {
            lpAioData = lpOverlappedEx->m_pAioData;
        }

        if( ( 0 == nRet ) && !bRet && lpAioData )
        {
            lpAioData->m_dwErrCode = dwErr;
        }

        if( ( 0 == nRet ) && !lpAioData )
        {
            cout << "iocp closed" << endl;
            nRet = 1;
        }

        if( 0 == nRet )
        {
            aio_helper tool( m_pAioImpl );
            tool.proc( bRet ? 0 : 1, dwBytes, lpAioKey, lpAioData );
        }

        if( 2 == nRet )
        {
            nRet = 0;
        }

        return nRet;
    }

    int aio_thread::proc_close()
    {
        int ret = 0;

        aio_data::LIST & data_list = m_pAioImpl->m_closing_list;

        for( aio_data::ITER iter = data_list.begin(); iter != data_list.end(); )
        {
            aio_data::ITER cur_iter = iter++;
            aio_data * pData = *cur_iter;
            xos_box::i_msg * pMsg = pData->m_lpMsgData;
            aio_key * pKey = (aio_key *)pMsg->get_void( 1, 0 );
            int num = pKey->post_request_num();
            if( 0 == num )
            {
                pKey->finish_send_request( m_pAioImpl );
                xos::i_socket * lpSocket = pKey->m_pSocket;
                int onoff = pMsg->get_int( 1, 0 );
                int linger = pMsg->get_int( 2, 0 );
                lpSocket->close( onoff, linger );
                pMsg->set_bool( 0, true );
                pMsg->set_void( 1, 0 );
                m_pAioImpl->notify( pMsg );
            }
            if( 0 == num )
            {
                xos_stl::release_interface( pKey );
                data_list.erase( cur_iter );
                pData->m_lpMsgData = 0;
                xos_stl::term_pool_item( pData );
            }
        }

        return ret;
    }

} // xos_aio
